package cn.dglydrpy.study.ssm.common.kafka.deserializer;

import java.nio.ByteBuffer;
import java.util.Map;

import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Deserializer;

import cn.dglydrpy.study.ssm.common.kafka.domain.Customer;
/**
 反序列化
	1) 创建自定义的反序列化器
	2) 更改值的反序列化器
	3) 更改消费者的值类型
	4) 更改接收到的消息的泛型类型
 * @author hasee
 *
 */
public class CustomerDeserializer implements Deserializer<Customer> {
    @Override
    public void configure(Map<String, ?> map, boolean b) {

    }

    @Override
    public Customer deserialize(String s, byte[] bytes) {
        int customerId;
        int nameSize;
        String customerName;
        try {
            if (null == bytes) {
                return null;
            }
            if (bytes.length < 8) {
                throw new SerializationException("消息长度不对");
            }
            ByteBuffer buffer = ByteBuffer.wrap(bytes);
            customerId = buffer.getInt();
            nameSize = buffer.getInt();
            byte[] nameBytes = new byte[nameSize];
            buffer.get(nameBytes);
            customerName = new String(nameBytes, "UTF-8");
            Customer customer = new Customer();
            customer.setId(customerId);
            customer.setCustomerName(customerName);
            return customer;
        } catch (Exception e) {
            e.printStackTrace();
            throw new SerializationException("从字节数组反序列化到对象失败:" + e);
        }
    }

    @Override
    public void close() {

    }
}
